package com.flute.haflute.jobbox.master;

import org.safehaus.uuid.UUIDGenerator;
import org.slf4j.*;

import com.flute.haflute.jobbox.base.InputResource;
import com.flute.haflute.jobbox.base.mapreduce.MRJobConf;

/**
 * 执行中任务的信息
 * 
 * 跟踪任务执行的状态等
 * 
 * 需完善、重构
 * 
 */
public class MRJob {
	private static Logger logger = LoggerFactory.getLogger(MRJob.class);
	private InputResource inputSource = null;
	private MRJobConf jobConf;
	private String jobId;
	private int numMapTasks;
	private int numReduceTasks;

	public MRJob(MRJobConf jobConf) throws Exception {
		logger.debug("Job Initializing");
		if (jobConf == null)
			throw new Exception("jobconf should not be null");

		this.jobConf = jobConf;

		this.jobId = UUIDGenerator.getInstance().generateTimeBasedUUID()
				.toString();
		jobConf.setJobId(this.jobId);
		this.numMapTasks = jobConf.getNumMapTasks();
		this.numReduceTasks = jobConf.getNumReduceTasks();

		Class<? extends InputResource> inputClass = jobConf
				.getInputSourceClass();
		inputSource = inputClass.newInstance();
		logger.debug("Job Tracker Initialized successsful");
	}

	@Override
	public boolean equals(Object obj) {
		if (this == obj)
			return true;
		if (obj == null)
			return false;
		if (getClass() != obj.getClass())
			return false;
		final MRJob other = (MRJob) obj;
		if (jobId == null) {
			if (other.jobId != null)
				return false;
		} else if (!jobId.equals(other.jobId))
			return false;
		if (numMapTasks != other.numMapTasks)
			return false;
		if (numReduceTasks != other.numReduceTasks)
			return false;
		return true;
	}

	public InputResource getInputSource() {
		return inputSource;
	}

	public MRJobConf getJobConf() {
		return jobConf;
	}

	public String getJobId() {
		return jobId;
	}

	public int getNumMapTasks() {
		return numMapTasks;
	}

	public int getNumReduceTasks() {
		return numReduceTasks;
	}

	@Override
	public int hashCode() {
		final int prime = 31;
		int result = 1;
		result = prime * result + ((jobId == null) ? 0 : jobId.hashCode());
		result = prime * result + numMapTasks;
		result = prime * result + numReduceTasks;
		return result;
	}

}
